package org.rcy.framework.rpc.registry.thrift;

import org.apache.thrift.ProcessFunction;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TBase;
import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageType;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolUtil;
import org.apache.thrift.protocol.TType;
import org.rcy.framework.api.exception.BaseException;
import org.rcy.framework.thrift.ReflectionUtils;
import org.rcy.framework.thrift.log.RpcLog;
import org.rcy.framework.thrift.metadata.MethodMetaData;
import org.rcy.framework.thrift.struct.ArgsStruct;
import org.rcy.framework.thrift.struct.ResultStruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class DynamicProcessor<I> implements TProcessor {
	
	private final Logger logger = LoggerFactory.getLogger(getClass());

	private final Class<?> iface;
	private final I impl;
	private final Map<String, ProcessFunction<I, ? extends TBase>> processMap = new HashMap();
	
	/**
	 * @param iface 接口类型，
	 * @param impl 接口实现类，必须是iface的子类
	 * @throws TException 
	 */
	public DynamicProcessor(Class<?> iface,I impl) throws TException {
		this.iface = iface;
		this.impl = impl;
		this.initProcessMap();
	}

	@Override
	public void process(TProtocol in, TProtocol out) throws TException {
		TMessage msg = in.readMessageBegin();
		//根据方法名获取处理类ProcessFunction
		ProcessFunction fn = processMap.get(msg.name);
		if (fn == null) {
			TProtocolUtil.skip(in, TType.STRUCT);
			in.readMessageEnd();
			TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '" + msg.name + "'");
			out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));
			x.write(out);
			out.writeMessageEnd();
			out.getTransport().flush();
		}
		fn.process(msg.seqid, in, out, impl);
	}

	public Map<String, ProcessFunction<I, ? extends TBase>> getProcessMapView() {
		return Collections.unmodifiableMap(processMap);
	}

	/*
	 * 初始化接口方法处理函数,暂不支持接口类中继承的父接口的方法
	 */
	protected void initProcessMap() throws TException {
		Method[] methods = iface.getDeclaredMethods();
		Set<String> methodSet = new HashSet<>();
		for(Method method:methods){
			
			if(method.isDefault()){//java 8中的Default方法
				continue;
			}
			logger.debug("init method process:{}",method);
			String methodName = method.getName();
			if(methodSet.contains(methodName)) {
				throw new RuntimeException("rpc register method ["+iface.getSimpleName()+"."+methodName+"] duplicated");
			}else {
				methodSet.add(methodName);
			}
			Type returnType = method.getGenericReturnType();
			MethodMetaData metadata = MethodMetaData.resolveMetaData(method);
			
			MethodHandler methodHandler = new MethodHandler<I>(metadata,iface.getName()){
				@Override
				public TBase getResult(Object handler, ArgsStruct args) throws TException {
					ResultStruct result = new ResultStruct(returnType);
					RpcLog logInfo = null;
					long t1 = System.currentTimeMillis();
					try {
						Object[] values = args.getValues();
						if(values != null && values.length > 0 && values[values.length -1] instanceof RpcLog) {
							logInfo = (RpcLog) values[values.length -1];
							RpcLog.recoverTrace(logInfo);
							Object[] newValues = new Object[values.length -1];
							System.arraycopy(values, 0, newValues, 0, values.length-1);
							values = newValues;
						}
						Object successObj = ReflectionUtils.invokeMethod(method, impl,values);
						result.setSuccess(successObj);
					} catch (Exception e) {
						//服务端抛出异常，先在服务端记录日志，然后序列化到客户端（只序列化基本信息）
						if(e instanceof BaseException) {
							logger.warn("Invoke error, impl:{}, method:{} cause:{}",impl,methodName,e.getMessage());
						}else if(e instanceof IllegalArgumentException){
							//服务端抛出异常，先在服务端记录日志，然后序列化到客户端（只序列化基本信息）
							logger.error("Invoke error, impl:{}, method:{} cause:{}",impl,methodName,e.getMessage());
						}else {
							logger.error("Invoke error, impl:{}, method:{} cause:{}",impl,methodName,e.getMessage(),e);
						}
						result.setException(e);
					}catch (Throwable e) {
						logger.error("Invoke error, throwable impl:{}, method:{} cause:{}",impl,methodName,e.getMessage(),e);
						result.setException(new RuntimeException("rpc server throwable error ", e));
					}finally {
						long cost = System.currentTimeMillis()-t1;
						if(cost > 15000L) {
							logger.warn("Invoke warn, impl:{}, method:{} cost {}",impl,methodName,cost);
						}
						RpcLog.removeTrace();
					}
					return result;
				}
			};
			
			processMap.put(methodName, methodHandler);
		}
	}

}